<?php

declare(strict_types=1);

namespace App\CustHub\Listener;

use Hyperf\Event\Annotation\Listener;
use Hyperf\Event\Contract\ListenerInterface;
use Nashgao\MQTT\Event\OnReceiveEvent;
use Psr\Log\LoggerInterface;
use function Hyperf\Coroutine\co;

#[Listener]
class MqttOnReceiveListener implements ListenerInterface
{
    public function listen(): array
    {
        return [
            OnReceiveEvent::class,
        ];
    }

    public function process(object $event): void
    {
        // event 里通常包含 topic, payload, qos, properties 等字段（以包实现为准）
        // 这里做一个稳健访问（若字段名不同请按实际类调整）
        try {
            $topic = $event->topic;
            $payload = $event->message ;
            $qos = $event->qos ?? 0;
            var_dump($event);
            /* 简单日志 */
            $logger = container()->get(LoggerInterface::class);
            $logger->info('MQTT received', ['topic' => $topic, 'qos' => $qos, 'payload_len' => strlen((string)$payload)]);

            // 重要：不要在 event listener 同步做重耗时操作，建议入队或用 go 协程
            co(function () use ($topic, $payload) {
                try {
                    // 例如：推送到 Hyperf 的异步 Job
                    // dispatch(new \App\Job\ProcessMqttMessage($topic, $payload));
                    // 或者直接调用处理类
                    // (示例中直接把 payload 解析并写日志)
                    $data = json_decode((string)$payload, true);
//                    var_dump($data);
                    // 你的业务处理
                } catch (\Throwable $e) {
                    // 处理异常：记录、告警或重试入死信队列
                    container()->get(LoggerInterface::class)->error('MQTT processing error: ' . $e->getMessage());
                }
            });
        }catch (\Throwable $e){
            var_dump($e->getMessage());
        }
    }
}